package com.bucg.cjcloud.gb28181.media.conf;

import com.bucg.cjcloud.gb28181.media.cache.ConcurrentHashMapCacheUtils;
import com.bucg.cjcloud.gb28181.media.event.MediaServerEvent;
import com.bucg.cjcloud.gb28181.media.event.MediaStatus;
import com.bucg.cjcloud.gb28181.sip.modle.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

// @Component
/**
 * 与媒体服务器心跳检测
 */
public class NIOEchoServer implements ApplicationRunner {



    private ByteBuffer buffer = ByteBuffer.allocate(128);

    private int TIME_OUT_MILLES = 15000;

    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    MediaProperties mediaProperties;


    private void startEchoServer() {

        Integer echoPort = mediaProperties.getEchoPort();
        if (echoPort != null && (echoPort < 1024 || echoPort >= 65536))
            throw new RuntimeException("echo port must be 1024~65536");

        final int DEFAULT_PORT = echoPort == null ? 1125 : echoPort;

        //open Selector and ServerSocketChannel by calling the open() method
        try (Selector selector = Selector.open();
             ServerSocketChannel serverSocketChannel = ServerSocketChannel.open()) {

            //check that both of them were successfully opened
            if ((serverSocketChannel.isOpen()) && (selector.isOpen())) {

                //configure non-blocking mode
                serverSocketChannel.configureBlocking(false);

                //set some options
                serverSocketChannel.setOption(StandardSocketOptions.SO_RCVBUF, 256 * 1024);
                serverSocketChannel.setOption(StandardSocketOptions.SO_REUSEADDR, true);

                //bind the server socket channel to port
                serverSocketChannel.bind(new InetSocketAddress(DEFAULT_PORT));

                //register the current channel with the given selector
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

                //display a waiting message while ... waiting!
                System.out.println("Waiting for connections ...");

                while (true) {
                    //wait for incomming events
                    selector.select();

                    //there is something to process on selected keys
                    Iterator keys = selector.selectedKeys().iterator();

                    while (keys.hasNext()) {
                        SelectionKey key = (SelectionKey) keys.next();

                        //prevent the same key from coming up again
                        keys.remove();

                        if (!key.isValid()) {
                            continue;
                        }

                        if (key.isAcceptable()) {

                            acceptOP(key, selector);
                        } else if (key.isReadable()) {
                            this.readOP(key);
                        } else if (key.isWritable()) {
                            this.writeOP(key);
                        }
                    }
                }
            } else {
                System.out.println("The server socket channel or selector cannot be opened!");
            }
        } catch (IOException ex) {
            System.err.println(ex);
        }
    }

    //isAcceptable returned true
    private void acceptOP(SelectionKey key, Selector selector) throws IOException {

        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
        SocketChannel socketChannel = serverChannel.accept();
        socketChannel.configureBlocking(false);

        System.out.println("Incoming connection from: " + socketChannel.getRemoteAddress());

        //register channel with selector for further I/O
        socketChannel.register(selector, SelectionKey.OP_READ);
    }

    //isReadable returned true
    private void readOP(SelectionKey key) {

        try {
            SocketChannel socketChannel = (SocketChannel) key.channel();

            buffer.clear();

            int numRead = -1;
            try {
                numRead = socketChannel.read(buffer);
            } catch (IOException e) {
                System.err.println("Cannot read error!");
            }

            if (numRead == -1) {
                System.out.println("Co" +
                        "" +
                        "" +
                        "nnection closed by: " + socketChannel.getRemoteAddress());
                socketChannel.close();
                key.cancel();
                return;

            }

            byte[] data = new byte[numRead];
            System.arraycopy(buffer.array(), 0, data, 0, numRead);


            // write back to client
            doEchoJob(key, data);
        } catch (IOException ex) {
            System.err.println(ex);
        }
    }

    //isWritable returned true
    private void writeOP(SelectionKey key) throws IOException {

        SocketChannel socketChannel = (SocketChannel) key.channel();

        Long aLong = (Long) ConcurrentHashMapCacheUtils.getCache((String) key.attachment());
        if (aLong != null)
            socketChannel.write(ByteBuffer.wrap(aLong.toString().getBytes()));

        key.interestOps(SelectionKey.OP_READ);
    }

    private void doEchoJob(SelectionKey key, byte[] data) throws IOException {

        SocketChannel socketChannel = (SocketChannel) key.channel();
        InetSocketAddress remoteAddress = (InetSocketAddress) socketChannel.getRemoteAddress();

        String rawData = new String(data, "UTF-8");
        String[] split = rawData.split(":");
        if (split.length < 2) return;
        String servicePort = (split[0]);
        Long nTick = Long.valueOf(split[1]);


        InetAddress address = remoteAddress.getAddress();
        String serveraddr = address + ":" + servicePort;

        Long cache = (Long) ConcurrentHashMapCacheUtils.getCache(serveraddr);
        cache = cache == null ? 0 : cache;
        if (cache == 0)
            applicationContext.publishEvent(new MediaServerEvent(new Pair<String,String>(address.toString(), servicePort), MediaStatus.ONLINE));

        if (nTick <= cache && cache<Long.MAX_VALUE ) {
            // Media 之前可能离线
            applicationContext.publishEvent(new MediaServerEvent(new Pair<>(address.toString(), servicePort), MediaStatus.ONCE));
        }
        ConcurrentHashMapCacheUtils.setCache(serveraddr, nTick, TIME_OUT_MILLES);

        key.attach(serveraddr);
        key.interestOps(SelectionKey.OP_WRITE);
    }

    public static void main(String[] args) {
        NIOEchoServer main = new NIOEchoServer();
        main.startEchoServer();
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        // EchoNonBlockingServer main = new EchoNonBlockingServer();
        startEchoServer();
    }


}

